package com.zx.notification.consumer.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zx.notification.common.enums.MqTypeEnum;
import com.zx.notification.common.event.ConfigReloadEvent;
import com.zx.notification.common.model.NotificationChannelsConfig;
import com.zx.notification.consumer.config.ChannelConfigManager;
import com.zx.notification.consumer.config.RabbitMqConsumerConfig;
import com.zx.notification.consumer.service.IMessageProcessService;
import com.zx.notification.consumer.util.MessageProcessUtil;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RabbitMQ消息监听器
 * 用于处理RabbitMQ通知消息
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RabbitMqMsgListener {

    private final IMessageProcessService messageProcessService;
    private final ChannelConfigManager channelConfigManager;
    private final RabbitMqConsumerConfig rabbitMqConfig;
    private final ObjectMapper objectMapper = new ObjectMapper();
    // 存储动态创建的监听器容器
    private final Map<String, MessageListenerContainer> listenerContainers = new ConcurrentHashMap<>();
    // 存储渠道类型与目标名称的映射关系
    private final Map<String, String> channelDestinationMap = new ConcurrentHashMap<>();
    // RabbitMQ连接工厂
    @Autowired
    private ConnectionFactory connectionFactory;

    /**
     * 初始化监听器，从数据库加载启用的RabbitMQ渠道配置
     */
    @PostConstruct
    public void init() {
        if (!rabbitMqConfig.isRabbitmqEnabled()) {
            log.info("RabbitMQ未启用，跳过RabbitMQ监听器初始化");
            return;
        }

        // 验证必要配置
        if (connectionFactory == null) {
            log.error("RabbitMQ连接工厂未配置，无法初始化监听器");
            return;
        }

        loadActiveRabbitMqChannelConfigurations();
    }

    /**
     * 优雅关闭，清理资源
     */
    @PreDestroy
    public void destroy() {
        log.info("开始关闭RabbitMQ监听器...");
        synchronized (listenerContainers) {
            listenerContainers.values().forEach(container -> {
                try {
                    container.stop();
                    if (container instanceof SimpleMessageListenerContainer) {
                        ((SimpleMessageListenerContainer) container).destroy();
                    }
                } catch (Exception e) {
                    log.error("关闭监听器容器失败", e);
                }
            });
            listenerContainers.clear();
            channelDestinationMap.clear();
        }
        log.info("RabbitMQ监听器关闭完成");
    }

    /**
     * 从数据库加载启用的RabbitMQ渠道配置
     */
    private void loadActiveRabbitMqChannelConfigurations() {
        try {
            // 查询所有启用的RabbitMQ渠道配置
            List<NotificationChannelsConfig> channelConfigs = channelConfigManager.getActiveChannelConfigsByMqType(MqTypeEnum.RABBITMQ.getCode());

            if (channelConfigs.isEmpty()) {
                log.info("未找到启用的RabbitMQ渠道配置");
                return;
            }

            for (NotificationChannelsConfig config : channelConfigs) {
                // 添加队列名称
                String queueName = config.getQueueName();
                if (queueName != null && !queueName.trim().isEmpty()) {
                    channelDestinationMap.put(config.getChannelType() + "_QUEUE", queueName);
                    registerListenerForQueue(queueName);
                }

                // 添加主题名称
                String topicName = config.getTopicName();
                if (topicName != null && !topicName.trim().isEmpty()) {
                    channelDestinationMap.put(config.getChannelType() + "_TOPIC", topicName);
                    registerListenerForQueue(topicName);
                }

                log.info("加载启用的RabbitMQ渠道配置: {} -> 队列:{}, 主题:{}",
                        config.getChannelType(), queueName, topicName);
            }
        } catch (Exception e) {
            log.error("加载启用的RabbitMQ渠道配置失败", e);
        }
    }

    /**
     * 为指定队列注册监听器
     *
     * @param queueName 队列名称
     */
    private void registerListenerForQueue(String queueName) {
        if (listenerContainers.containsKey(queueName)) {
            log.debug("监听器已存在: {}", queueName);
            return;
        }

        int maxRetries = 3;
        for (int attempt = 1; attempt <= maxRetries; attempt++) {
            try {
                SimpleMessageListenerContainer container = createMessageListenerContainer(queueName);
                // 启动监听器容器
                container.start();

                // 存储监听器容器引用，便于后续管理
                listenerContainers.put(queueName, container);

                log.info("成功为队列 {} 注册监听器 (尝试 {}/{})", queueName, attempt, maxRetries);
                return;

            } catch (Exception e) {
                log.error("为队列 {} 注册监听器失败 (尝试 {}/{}): {}", queueName, attempt, maxRetries, e.getMessage());

                if (attempt == maxRetries) {
                    log.error("为队列 {} 注册监听器最终失败，已重试 {} 次", queueName, maxRetries, e);
                } else {
                    try {
                        // 重试前等待
                        Thread.sleep(1000L * attempt);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        log.error("重试等待被中断", ie);
                        break;
                    }
                }
            }
        }
    }

    /**
     * 创建消息监听器容器
     *
     * @param queueName 队列名称
     * @return 消息监听器容器
     */
    private SimpleMessageListenerContainer createMessageListenerContainer(String queueName) {
        // 创建消息监听器容器
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames(queueName);
        container.setConcurrentConsumers(1); // 设置并发消费者数量

        // 创建消息监听器适配器
        MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(this, "handleSingleMessage");
        container.setMessageListener(listenerAdapter);

        // 设置异常处理
        container.setErrorHandler(t -> {
            log.error("监听器容器发生未处理异常，队列: {}", queueName, t);
        });

        return container;
    }

    /**
     * 处理单个RabbitMQ通知消息
     *
     * @param message 通知消息（JSON格式字符串）
     */
    public void handleSingleMessage(String message) {
        if (!rabbitMqConfig.isRabbitmqEnabled()) {
            log.debug("RabbitMQ未启用，忽略接收到的消息");
            return;
        }

        if (message == null || message.trim().isEmpty()) {
            log.warn("接收到空消息，忽略处理");
            return;
        }

        long startTime = System.currentTimeMillis();
        try {
            MessageProcessUtil.processMessage(message, MqTypeEnum.RABBITMQ, objectMapper, channelConfigManager, messageProcessService);
            long processingTime = System.currentTimeMillis() - startTime;
            log.debug("消息处理完成，耗时: {}ms", processingTime);
        } catch (Exception e) {
            long processingTime = System.currentTimeMillis() - startTime;
            log.error("处理RabbitMQ消息失败，耗时: {}ms", processingTime, e);
            // 根据业务需求决定是否抛出异常
        }
    }

    /**
     * 重新加载渠道配置
     */
    public void reloadChannelConfigurations() {
        if (!rabbitMqConfig.isRabbitmqEnabled()) {
            log.debug("RabbitMQ未启用，跳过重新加载渠道配置");
            return;
        }

        log.info("开始重新加载RabbitMQ渠道配置...");

        // 使用同步块防止并发重载
        synchronized (listenerContainers) {
            // 停止所有现有的监听器容器
            listenerContainers.values().forEach(container -> {
                try {
                    container.stop();
                    if (container instanceof SimpleMessageListenerContainer) {
                        ((SimpleMessageListenerContainer) container).destroy();
                    }
                } catch (Exception e) {
                    log.error("停止监听器容器失败", e);
                }
            });
            listenerContainers.clear();

            channelDestinationMap.clear();
            loadActiveRabbitMqChannelConfigurations();
        }

        log.info("RabbitMQ渠道配置重新加载完成");
    }

    /**
     * 监听配置重载事件
     *
     * @param event 配置重载事件
     */
    @EventListener
    public void handleConfigReloadEvent(ConfigReloadEvent event) {
        if (!rabbitMqConfig.isRabbitmqEnabled()) {
            log.debug("RabbitMQ未启用，忽略配置重载事件");
            return;
        }

        log.info("收到配置重载事件: {}", event.getMessage());

        // 延迟执行，避免频繁重载
        try {
            // 1秒延迟
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("配置重载延迟被中断", e);
            return;
        }

        // 重新加载渠道配置
        reloadChannelConfigurations();
    }
}